package br.ufmg.dcc.vod.remoteworkers.scheduler;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;

import br.edu.ufcg.lsd.commune.api.FailureNotification;
import br.edu.ufcg.lsd.commune.api.InvokeOnDeploy;
import br.edu.ufcg.lsd.commune.api.MonitoredBy;
import br.edu.ufcg.lsd.commune.api.RecoveryNotification;
import br.edu.ufcg.lsd.commune.container.servicemanager.ServiceManager;
import br.edu.ufcg.lsd.commune.identification.ServiceID;
import br.ufmg.dcc.vod.remoteworkers.Command;
import br.ufmg.dcc.vod.remoteworkers.ID;
import br.ufmg.dcc.vod.remoteworkers.Result;
import br.ufmg.dcc.vod.remoteworkers.processor.Processor;
import br.ufmg.dcc.vod.remoteworkers.scheduler.ProcessorEntry.ProcessorState;

public class Scheduler implements CommandInterested {

	public static final String OBJECT_NAME = "SCHEDULER";
	
	private ServiceManager serviceManager;
	
	private final Map<ID, String> commandIDtoProcessor;
	private final Map<String, ProcessorEntry> processors;
	private final Map<String, Long> tokens;

	private SchedulerListener listener;

	public Scheduler() {
		this.commandIDtoProcessor = new HashMap<ID, String>();
		this.processors = new HashMap<String, ProcessorEntry>();
		this.tokens = new HashMap<String, Long>();
		
	}
	
	@InvokeOnDeploy
	public void init(ServiceManager serviceManager) {
		this.serviceManager = serviceManager;
	}
	
	public void setProcessors(Set<ServiceID> sids) {
		for (ServiceID sid : sids) {
			this.processors.put(sid.toString(), new ProcessorEntry());
			this.tokens.put(sid.toString(), Processor.IDLE_TOKEN);
			this.serviceManager.registerInterest(OBJECT_NAME, sid.toString(), Processor.class);
		}
	}
	
	public void setListener(SchedulerListener listener) {
		this.listener = listener;
	}
	
	@Override
	public void commandException(ID id, Exception e) {
		String sid = this.commandIDtoProcessor.remove(id);
		if (processors.containsKey(sid)) {
			this.processors.get(sid).setIdle();
			this.processors.get(sid).getInstance().release(this.tokens.remove(sid).longValue());
			this.listener.dealWithCommandException(id, e);
		}
	}

	@Override
	public void commandResult(ID id, Result result) {
		String sid = this.commandIDtoProcessor.remove(id);
		
		if (processors.containsKey(sid)) {
			this.processors.get(sid).setIdle();
			this.processors.get(sid).getInstance().release(this.tokens.remove(sid).longValue());
			this.listener.dealWithCommandResult(id, result);
		}
	}

	@Override
	public void iAmWorking(long token, @MonitoredBy(OBJECT_NAME) Processor processor) {
		String sid = serviceManager.getStubDeploymentID(processor).getServiceID().toString();
		if (processors.containsKey(sid)) {
			this.tokens.put(sid, token);
		}
	}

	@FailureNotification
	public void processorDown(@MonitoredBy(OBJECT_NAME) Processor processor) {
		String sid = serviceManager.getStubDeploymentID(processor).getServiceID().toString();
		
		if (processors.containsKey(sid)) {
			ProcessorState oldState = this.processors.get(sid).getState();
			Command command = this.processors.get(sid).getCurrentCommand();
			
			this.processors.get(sid).setDown();
			if (oldState == ProcessorState.WORKING) {
				this.listener.dealWithCommandException(command.getID(), new ProcessorDiedException());
			}
		}
	}
	
	@RecoveryNotification
	public void processorUp(@MonitoredBy(OBJECT_NAME) Processor processor) {
		
		String sid = serviceManager.getStubDeploymentID(processor).getServiceID().toString();
		
		if (processors.containsKey(sid)) {
			this.processors.get(sid).setIdle();
			this.processors.get(sid).setInstance(processor);
			this.listener.idleProcessorAvailable();
		}	
	}
	
	public ServiceID fetchAvailableProcessorID() {
		for (Entry <String, ProcessorEntry> e : processors.entrySet()) {
			String sid = e.getKey();
			if (e.getValue().getState() == ProcessorState.IDLE) {
				return ServiceID.parse(sid);
			}
		}
		
		return null;
	}
	
	public void dispatchCommand(ServiceID sid, Command command) {
		if (this.commandIDtoProcessor.containsKey(command.getID())) {
			throw new RuntimeException("Already executing a command with this id " + command.getID());
		}
			
		if (!processors.containsKey(sid.toString())) {
			throw new RuntimeException("Unknown id " + sid);
		}
		
		ProcessorEntry entry = this.processors.get(sid.toString());
		entry.setWorking(command);
		Processor proc = entry.getInstance();
		proc.process(this, command);
		commandIDtoProcessor.put(command.getID(), sid.toString());
	}
}